import { APIClientBase } from "cluster-client";
import DineggRegistryClient from "./DineggRegistryClient";
import * as crypto from "crypto";
import { EggApplication } from "egg";

/** 用于进程间socket直连通讯的封装类 */
export default class DineggAPIClient extends APIClientBase {
	constructor(private options: { isAgent: boolean; app: EggApplication; [key: string]: any }) {
		super(options);
		this.isAgent = !!options.isAgent;
	}

	private isAgent: boolean = false;

	/** 多进程通讯客户端 */
	private _client: DineggRegistryClient;

	/** 保存cluster的 app pid列表 */
	public appPids: Set<number> = new Set();
	public agentPid: number | null = null;
	/** app pids迭代器 */
	private appPidsIterator: IterableIterator<number> | null = null;

	// @ts-ignore 返回原始的客户端类
	private get DataClient() {
		return DineggRegistryClient;
	}

	// @ts-ignore 用于设置 cluster-client 相关参数，等同于 cluster 方法的第二个参数
	private get clusterOptions() {
		return {
			responseTimeout: 60 * 1000,
		};
	}

	/** 初始化订阅 */
	private initSubscribe() {
		// 发送自己的pid给到其他触发此事件的进程
		this.subscribe("get-pids", () => {
			if (this.isAgent) {
				this.publish("send-my-pid", { isAgent: true, pid: process.pid });
			} else {
				this.publish("send-my-pid", { isAgent: false, pid: process.pid });
			}
		});
		this.subscribe("send-my-pid", (msg: { pid: number; isAgent: boolean }) => {
			// console.log("send my pid", msg);
			if (msg.isAgent) {
				this.agentPid = msg.pid;
			} else {
				this.appPids.add(msg.pid);
			}
		});

		// 特殊用于fetch请求时来响应
		this.subscribe("fetch-response", (msg: { messageId: string; data: any }) => {
			if (!msg.messageId) return;
			this.emit(msg.messageId, msg.data);
		});
	}

	/** 订阅一个事件 */
	subscribe<T extends any>(event: string, listener: (data: T) => void) {
		this._client.subscribe({ dataId: event }, listener);
	}

	/** 发布消息 */
	publish(event: string, data: any) {
		this._client.publish({ dataId: event, data: data });
	}

	/**
	 *  请求响应模型
	 * - 使用publish发布一个事件，发布时携带一个唯一messageId，
	 * - 某个订阅端执行完成后，再以此messageId 来publish事件
	 * - 由于app worker有多个，agent只一个，应该只用在app中，否则如果在agent上用，可能有多个app接收并处理逻辑
	 * - agent使用publish配合使用，或使用fetchResponse方法配合
	 */
	async fetch<T>(event: string, data: any, opt?: { timeout?: number }): Promise<T> {
		const messageId = `${process.pid}_${event}_${crypto.randomBytes(12).toString("hex")}`;
		return new Promise((resolve, reject) => {
			const timeout = opt?.timeout || 1000 * 12;
			const timer = setTimeout(() => {
				reject(new Error(`your fetch event [${event}] is timeout.`));
			}, timeout);
			this.once(messageId, (data) => {
				resolve(data);
				clearTimeout(timer);
			});
			this.publish(event, {
				event,
				messageId,
				data,
			});
		});
	}

	/**
	 * 一般用于agent端响应fetch过来的请求
	 */
	fetchResponse(messageId: string, data: any) {
		this.publish("fetch-response", { messageId, data });
	}

	/** 进程监听只自己接收的订阅，可实现一个事件只一个进程接收到消息 */
	onlyMeSubscribe<T extends any>(event: string, listener: (data: T) => void) {
		event = `${process.pid}.${event}`;
		this.subscribe(event, listener);
	}

	/** 发送publish到某个进程，不区分app还是agent，只要知道pid即可 */
	publishTo(pid: number, event: string, data: any) {
		event = `${pid}.${event}`;
		this.publish(event, data);
	}

	/**
	 * - agent 上调用会直接触发自身订阅
	 * - app 上调用，发送给agent
	 */
	sendToAgent(event: string, data: any) {
		if (!this.agentPid) {
			throw new Error(`can not get agent process id.`);
		}
		this.publishTo(this.agentPid, event, data);
	}

	/**
	 * - 发送给某个app进程
	 * - agent上发送给一个app 按轮询调度所有app worker
	 * - app上直接触发自身订阅
	 */
	sendToApp(event: string, data: any) {
		if (this.isAgent) {
			const pid = this.getAppPid();
			this.publishTo(pid, event, data);
		} else {
			// app进程，触发本进程订阅
			this.publishTo(process.pid, event, data);
		}
	}

	/** 迭代器获取app pid列表 */
	private getAppPid() {
		const size = this.appPids.size;
		if (size < 1) throw new Error(`can not get app pids. please manual get it.`);
		if (!this.appPidsIterator) {
			this.appPidsIterator = this.appPids.values();
		}
		let pid = this.appPidsIterator.next();
		if (pid.done) {
			this.appPidsIterator = this.appPids.values();
			pid = this.appPidsIterator.next();
			if (pid.done) throw new Error(`the app pid pools is empty.`);
			return pid.value;
		} else {
			return pid.value;
		}
	}

	/**  等待初始化连接完成 */
	async ready() {
		await this._client.ready();
		// await super.ready(...args);
		// 连接ready后，再订阅内置消息事件
		setTimeout(() => {
			this.initSubscribe();
		}, 2000);
		setTimeout(() => {
			// console.log(process.pid, `ready，sync cluster pids.`);
			this.syncClusterPids();
		}, 5000);
		// console.log("init dineggApiClient ready ", process.pid);
		this.options.app.coreLogger.info(`init dineggApiClient ready %s`, process.pid);
	}

	/** 声明父类ts类型 */
	once(event: string, listener: any) {
		super.once(event, listener);
	}
	/** 声明父类ts类型 */
	on(event: string, listener: any) {
		super.on(event, listener);
	}

	emit(event: string, data: any) {
		super.emit(event, data);
	}

	/** 触发集群，让所有进程发送自己的pid给所有人。相当于每个进程都更新了自己维护的pid列表 */
	syncClusterPids() {
		this.appPids = new Set();
		this.publish("get-pids", 0);
	}
}
